package com.study.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.{SparkConf, SparkContext}

object Scala20_RDD_Operator_Transform{
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    conf.set("spark.local.dir","D:\\hadoopbook\\spark\\test")
    val sc = new SparkContext(conf)
    // TODO 算子 - 转换 -  join
    val rdd = sc.makeRDD(List(("a",1),("b",2),("c",3)))
    val rdd1 = sc.makeRDD(List(("a",4),("d",5),("c",6)))
    // spark中join主要针对两个数据集之间相同key的数据连接
    // 可能出现笛卡尔乘积，可能会出现shuffle，性能差。尽量不使用
    val rdd2 = rdd.join(rdd1)
    rdd2.collect().foreach(println)
    /*
    (a,(1,4))
    (c,(3,6))
     */
    val rdd3 = rdd.leftOuterJoin(rdd1)
    rdd3.collect().foreach(println)
    /*
    (a,(1,Some(4)))
    (b,(2,None))
    (c,(3,Some(6)))
     */
    val rdd4 = rdd.rightOuterJoin(rdd1)
    rdd4.collect().foreach(println)
    /*
    (a,(Some(1),4))
    (c,(Some(3),6))
    (d,(None,5))
     */
    val rdd5 = rdd.fullOuterJoin(rdd1)
    rdd5.collect().foreach(println)
    /*
    (a,(Some(1),Some(4)))
    (b,(Some(2),None))
    (c,(Some(3),Some(6)))
    (d,(None,Some(5)))
     */
    val rdd7 = sc.makeRDD(List(("a",1),("b",2),("c",3)))
    val rdd8 = sc.makeRDD(List(("a",4),("a",5),("b",6)))
    // connect + group
    val rdd6 = rdd7.cogroup(rdd8)
    rdd6.collect().foreach(println)
    /*
    (a,(CompactBuffer(1),CompactBuffer(4, 5)))
    (b,(CompactBuffer(2),CompactBuffer(6)))
    (c,(CompactBuffer(3),CompactBuffer()))
     */
    sc.stop()

  }

}
